package book.data_clean;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.log4j.BasicConfigurator;

import java.io.IOException;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class cleanMR {
    public static class CleanMapper extends Mapper<LongWritable, Text, Text, NullWritable> {
        Data k = new Data();

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

            //解决编码问题
            String line = new String(value.getBytes(), 0, value.getLength(), "GBK");
            //去除数据的第一行也就是列名
            if (line.startsWith("positionName")) return;
            //将去除了列名行的数据以分割符分割
            //因为有些列里面有分割符‘，’
            String[] data = line.split(",(?=(?:[^\"]*\"[^\"]*\")*[^\"]*$)", -1);
            //判断是否有空值，有的话删除
            for (String d : data) {
                if (d == null || d.equals("")) return;
            }
            //判断倒数第二列是否是字母乱码
            //倒数第二列应该是数字，不能有字母、符号、汉字
            Pattern pattern = Pattern.compile("[0-9]*");
            Matcher matcher = pattern.matcher(data[data.length - 2]);
            if (!matcher.matches()) return;
            //将薪资中的9k-10k*2改为9k,10k俩列
            String minSalary = data[1].split("-")[0];
            String maxSalary;
            //如果包含错误符号*
            if (data[1].contains("*")){
                maxSalary = data[1].split("-")[1].split("\\*")[0];
            }else {
                maxSalary = data[1].split("-")[1];
            }
            //将数据装载如JavaBean
            k.set(data[0], minSalary + "-" + maxSalary, Integer.parseInt(minSalary.substring(0, minSalary.length() - 1)),
                    Integer.parseInt(maxSalary.substring(0, maxSalary.length() - 1)), data[2], data[3], data[4], data[5], data[6], data[7], data[8], data[9], Integer.parseInt(data[10]), Integer.parseInt(data[11]), data[12]);            //将数据输出
            context.write(new Text(k.toString()), NullWritable.get());
        }
    }

    public static class ClearReducer extends Reducer<Text, NullWritable, Text, NullWritable> {
        @Override
        protected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
            context.write(key, NullWritable.get());
        }
    }

    public static void main(String[] args) {
        //日志
        BasicConfigurator.configure();
        Configuration conf = new Configuration();
        try {
            //配置环境
            conf.set("fs.default", "hdfs://192.168.0.155:9000/");
            Job job = Job.getInstance(conf);
            //设置driver、mapper和driver类
            job.setJarByClass(CleanDriver.class);
            job.setMapperClass(book.data_clean.CleanMapper.class);
            job.setReducerClass(book.data_clean.ClearReducer.class);

            //设置map输出的key和value的数据类型
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(NullWritable.class);

            //设置map输出的key和value的数据类型
            job.setOutputKeyClass(Text.class);
            job.setOutputKeyClass(NullWritable.class);

            //文件的读取路径
            FileInputFormat.setInputPaths(job, "/MR/input");

            //结果的数据路径
            Path path = new Path("/MR/out");
            //如果路径存在那么就删除
            if (path.getFileSystem(conf).exists(path)) {
                path.getFileSystem(conf).delete(path, true);
            }
            FileOutputFormat.setOutputPath(job, path);
            System.out.println(job.waitForCompletion(true) ? 0 : 1);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }
    }
}
